package actor.atLeastoncedelivery

import actor.atLeastoncedelivery.CalcAggregator._
import akka.actor.{ActorLogging, ActorRef, Props}
import akka.persistence._
import akka.persistence.AtLeastOnceDelivery.{AtLeastOnceDeliverySnapshot, UnconfirmedWarning}

import scala.util.control.NoStackTrace

/**
 *  至少发送一次
 * Created by DJJ on 2019/12/29.
 */
object CalcAggregator {

  sealed trait Command

  case class Add(x: Int, y: Int) extends Command

  case class Sub(x: Int, y: Int) extends Command

  case class Mul(x: Int, y: Int) extends Command

  case class Div(x: Int, y: Int) extends Command

  case class Result(id: Long, res: Int) extends Command

  case class Snap(results: List[Int], deliverySnapshot: AtLeastOnceDeliverySnapshot)

  case object ShowResults

  case object Boom

  case object ClearJournal

  def props(calculators: Map[String, ActorRef], keepJournalNr: Int) =
    Props(new CalcAggregator(calculators, keepJournalNr))
}

class CalcAggregator(calculators: Map[String, ActorRef], keepJournalNr: Int) extends PersistentActor with AtLeastOnceDelivery with ActorLogging {
  var results: List[Int] = List()
  var resultsId: Int = 0

  override def persistenceId: String = "calculation-actor023"

  //sending commands and update state only with delivery ack
  def updateState(cmd: Command) = {
    if (!recoveryRunning && !cmd.isInstanceOf[Result])
      log.info(s"Sending command message: $cmd at: $lastSequenceNr")

    cmd match {
      case Add(x, y) => deliver(calculators("ADD").path) { id => Calculator.Add(id, x, y) }
      case Sub(x, y) => deliver(calculators("SUB").path) { id => Calculator.Sub(id, x, y) }
      case Mul(x, y) => deliver(calculators("MUL").path) { id => Calculator.Mul(id, x, y) }
      case Div(x, y) => deliver(calculators("DIV").path) { id => Calculator.Div(id, x, y) }
      case Result(id, res) =>
        log.info(s"Receive calculation result $res with ack id: $id")
        if (res != 0) {
          results = res :: results
          confirmDelivery(id)
          log.info(s"Current state updated to: $results at $lastSequenceNr")
          resultsId += 1
          if (resultsId % keepJournalNr == 0) {
            resultsId = 0
            saveSnapshot(Snap(results, getDeliverySnapshot))
            println("--------------")
            log.info(s"Saving snapshot with state $results, snapshot: $getDeliverySnapshot")
          }
        }
    }
  }


  override def receiveCommand: Receive = {
    case cmd: Command => persist(cmd) {
      updateState
    }
    case ack: Calculator.Ack =>
      updateState(Result(ack.id, ack.x))

    case UnconfirmedWarning(unconfirmedDeliveries) => //cancel outstanding message
      log.info(s"UnconfirmedWarning: $unconfirmedDeliveries ...")
      unconfirmedDeliveries.foreach { u =>
        log.info(s"Cancelling unconfirmedDeliveris $u")
        confirmDelivery(u.deliveryId)
      }

    case SaveSnapshotSuccess(m) =>
      log.info(s"Sucessfull saving snapshot: ${m} at: $lastSequenceNr")
      //clear journal and snapshot
      deleteSnapshots(SnapshotSelectionCriteria(maxSequenceNr = m.sequenceNr - 1))
      deleteMessages(m.sequenceNr)
    case SaveSnapshotFailure(m, cause) =>
      log.info(s"Saving snapshot failed because: ${cause}")
    case DeleteMessagesSuccess(toSeq) =>
      log.info(s"Succefull deleting journal upto: $toSeq")
    case DeleteMessagesFailure(cause, toSeq) =>
      log.info(s"Failed to delete journal upto: $toSeq because: $cause")
    case DeleteSnapshotsSuccess(crit) =>
      log.info(s"Successful delete snapshots for $crit")
    case DeleteSnapshotSuccess(m) =>
      log.info(s"Successful delete snapshot upto: ${m.sequenceNr}")
    case DeleteSnapshotsFailure(crit, cause) =>
      log.info(s"Failed to delete snapshots $crit because: $cause")
    case DeleteSnapshotFailure(m, cause) =>
      log.info(s"Failed to delete snapshot upto: ${m.sequenceNr} because: $cause")

    case ShowResults =>
      log.info(s"Show Current State: $results and lastSequenceNr : $lastSequenceNr")

    case "TakeSnapshot" =>
      log.info(s"Saving snapshot with state: $results ...")
      saveSnapshot(Snap(results, getDeliverySnapshot))

    case Boom =>
      log.info("Boom!")
      throw new RuntimeException("boom") with NoStackTrace
    case ClearJournal =>
      deleteMessages(lastSequenceNr)
      deleteSnapshots(SnapshotSelectionCriteria(maxSequenceNr = lastSequenceNr))

  }

  override def receiveRecover: Receive = {
    case cmd: Command => updateState(cmd)
      log.info(s"Replaying command: $cmd")
    case SnapshotOffer(md, snap: Snap) =>
      log.info(s"Loading snapshot at: ${md.sequenceNr} with state: ${snap.results}")
      results = snap.results
      setDeliverySnapshot(snap.deliverySnapshot)
      log.info(s"Updated state to $results with snapshot")
    case RecoveryCompleted =>
      log.info(s"Recovery compeleted with State: $results and lastSequenceNr=$lastSequenceNr")
  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Aggregator restarting with reason: ${reason.getMessage}")
    super.preRestart(reason, message)
  }

  override def warnAfterNumberOfUnconfirmedAttempts = 1


}
